package cn.spark.study.sql

import org.apache.spark.SparkContext
import org.apache.spark.sql.SQLContext
import java.util.Arrays
import org.apache.spark.sql.types.StructField
import org.apache.spark.sql.types.StringType
import org.apache.spark.sql.types.IntegerType
import org.apache.spark.sql.types.StructType
import org.apache.spark.sql.Row

object JSONDataSouceScala {
  def main(args: Array[String]): Unit = {
    val sc = new ContextUtil().getContext("JSONDataSouceScala");
    val sqlC = new SQLContext(sc)
     goodStudentinfoScore(sc,sqlC)

  }
  
  def goodStudentinfoScore(sc:SparkContext,sqlC:SQLContext){
    sqlC.read.json("hdfs://spark1:9000/spark-study/students.json").registerTempTable("students");
    val goodStudentDF = sqlC.sql("select name,score from students where score >= 80");
    
    val goodStudentScores = goodStudentDF.rdd.map { _(0) }.collect()
    
    val studentInfoJSON = Array("{\"name\":\"Leo\",\"age\":18}","{\"name\":\"Marry\",\"age\":17}","{\"name\":\"Jack\",\"age\":19}")
    //val structField = StructType(Array(StructField("name",StringType,true),StructField("age",IntegerType,true)))
    
    sqlC.read.json(sc.parallelize(studentInfoJSON, 3)).registerTempTable("student_info")
    var sql = "select name,age from student_info where name in ("
    
    for(i <- 0 until goodStudentScores.length ){
        sql += "'" + goodStudentScores(i) + "'"
        if(i < goodStudentScores.length - 1 ){
          sql += ","
        }
    }
    sql += ")"
    val  goodStudentInfoJSONDF = sqlC.sql(sql)
    
    val goodStudentRDD = goodStudentDF.rdd.map { row => (row.getAs[String]("name"), row.getAs[Long]("score")) }
        .join(goodStudentInfoJSONDF.rdd.map { row => (row.getAs[String]("name"),row.getAs[Long]("age")) })
        .map(tuple => Row(tuple._1,tuple._2._1.toInt,tuple._2._2.toInt))
        
   val structField = StructType(Array(
       StructField("name",StringType,true),
       StructField("score",IntegerType,true),
       StructField("age",IntegerType,true)))
   sqlC.createDataFrame(goodStudentRDD, structField).write.format("json").save("hdfs://spark1:9000/spark-study/good-students-scala")
   
  }
}









